﻿using System;
using System.Collections.Generic;
using System.Threading;
using System.Linq;
using System.Text;
using TwitterMatic.Shared.Data;
using TwitterMatic.Shared.Domain;
using Microsoft.Samples.ServiceHosting.AspProviders;
using System.Web.Security;
using TwitterMatic.Shared;
using LinqToTwitter;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.StorageClient;
using System.Diagnostics;
using System.Net;
using Microsoft.WindowsAzure.Diagnostics;

namespace TwitterMatic_WorkerRole
{
    public class WorkerRole : RoleEntryPoint
    {
        protected IConfigurationProvider configuration;
        protected ITimedTweetRepository repository;

        protected IConfigurationProvider Configuration
        {
            get
            {
                if (configuration == null)
                {
                    configuration = new AzureConfigurationProvider();
                }
                return configuration;
            }
            set
            {
                configuration = value;
            }
        }

        protected ITimedTweetRepository Repository
        {
            get
            {
                if (repository == null)
                {
                    CloudStorageAccount cloudStorageAccount = CloudStorageAccount.FromConfigurationSetting("StorageConnection");
                    CloudTableClient tableClient = cloudStorageAccount.CreateCloudTableClient();
                    tableClient.CreateTableIfNotExist("TimedTweet");

                    repository = new TimedTweetRepository(tableClient);
                }
                return repository;
            }
            set
            {
                repository = value;
            }
        }

        protected Thread enqueueingThread;
        protected Thread publishingThread;

        public override void Run()
        {
            Trace.WriteLine("Started TwitterMatic worker process.", "Information");

            Trace.WriteLine("Creating enqueueing thread...", "Information");
            enqueueingThread = new Thread(new ThreadStart(EnqueueUpdates));
            Trace.WriteLine("Created enqueueing thread.", "Information");

            Trace.WriteLine("Creating publishing thread...", "Information");
            publishingThread = new Thread(new ThreadStart(PublishUpdates));
            Trace.WriteLine("Created publishing thread.", "Information");

            Trace.WriteLine("Starting worker threads...", "Information");
            enqueueingThread.Start();
            publishingThread.Start();
            Trace.WriteLine("Started worker threads.", "Information");

            enqueueingThread.Join();
            publishingThread.Join();

            Trace.WriteLine("Stopped worker threads.", "Information");

            Trace.WriteLine("Stopped TwitterMatic worker process.", "Information");
        }

        protected void EnqueueUpdates()
        {
            CloudStorageAccount cloudStorageAccount = CloudStorageAccount.FromConfigurationSetting("StorageConnection");
            CloudQueueClient queueClient = cloudStorageAccount.CreateCloudQueueClient();
            CloudQueue updateQueue = queueClient.GetQueueReference("updatequeue");
            updateQueue.CreateIfNotExist();

            while (true)
            {
                Trace.WriteLine("[Enqueue] Checking for due tweets...", "Information");
                List<TimedTweet> dueTweets = Repository.RetrieveDue(DateTime.Now.ToUniversalTime());
                if (dueTweets.Count > 0)
                {
                    Trace.WriteLine("[Enqueue] " + dueTweets.Count.ToString() + " due tweets.", "Information");

                    foreach (var tweet in dueTweets)
                    {
                        if (tweet.SendStatus != "Pending delivery")
                        {
                            updateQueue.AddMessage(new CloudQueueMessage(tweet.RowKey));
                            tweet.SendStatus = "Pending delivery";
                            Repository.Update(tweet);
                            Trace.WriteLine("[Enqueue] Enqueued tweet " + tweet.RowKey + " for publishing.", "Information");
                        }
                    }
                    Trace.WriteLine("[Enqueue] Finished processing due tweets.", "Information");
                }
                else
                {
                    Trace.WriteLine("[Enqueue] No due tweets.", "Information");
                }
                Thread.Sleep(120000);
            }
        }

        protected void PublishUpdates()
        {
            CloudStorageAccount cloudStorageAccount = CloudStorageAccount.FromConfigurationSetting("StorageConnection");
            CloudQueueClient queueClient = cloudStorageAccount.CreateCloudQueueClient();
            CloudQueue updateQueue = queueClient.GetQueueReference("updatequeue");
            updateQueue.CreateIfNotExist();

            while (true)
            {
                Trace.WriteLine("[Publish] Checking for pending tweets...", "Information");
                while (updateQueue.PeekMessage() != null)
                {
                    CloudQueueMessage queueItem = null;
                    try
                    {
                        queueItem = updateQueue.GetMessage(new TimeSpan(0, 2, 0));
                    }
                    catch
                    {
                        continue;
                    }

                    Trace.WriteLine("[Publish] Preparing to send pending message " + queueItem.AsString + "...", "Information");

                    TimedTweet tweet = null;
                    try
                    {
                        tweet = Repository.RetrieveById("", queueItem.AsString);
                    }
                    finally
                    {
                        if (tweet == null)
                        {
                            Trace.WriteLine("[Publish] Pending message " + queueItem.AsString + " has been deleted. Cancelling publish...", "Information");
                            updateQueue.DeleteMessage(queueItem);
                        }
                    }
                    if (tweet == null)
                        continue;

                    IOAuthTwitter oAuthTwitter = new OAuthTwitter();
                    oAuthTwitter.OAuthConsumerKey = Configuration.ReadSetting("OAuthConsumerKey");
                    oAuthTwitter.OAuthConsumerSecret = Configuration.ReadSetting("OAuthConsumerSecret");
                    oAuthTwitter.OAuthToken = tweet.Token;
                    oAuthTwitter.OAuthTokenSecret = tweet.TokenSecret;

                    TwitterContext ctx = new TwitterContext(oAuthTwitter);

                    bool retryRequested = false;
                    try
                    {
                        if (!string.IsNullOrEmpty(ctx.UpdateStatus(tweet.Status).ID))
                        {
                            Trace.WriteLine("[Publish] Published tweet " + tweet.RowKey + ".", "Information");

                            tweet.SentOn = DateTime.Now;
                            tweet.SendStatus = "Published";
                            tweet.RetriesLeft = 0;
                            updateQueue.DeleteMessage(queueItem);
                            Repository.Update(tweet);
                        }
                        else
                        {
                            retryRequested = true;
                        }
                    }
                    catch
                    {
                        retryRequested = true;
                    }
                    finally
                    {
                        if (retryRequested)
                        {
                            tweet.RetriesLeft--;
                            if (tweet.RetriesLeft > 0)
                            {
                                tweet.SendStatus = "Retrying";

                                Trace.WriteLine("[Publish] Error publishing tweet " + tweet.RowKey + ". Retries left: " + tweet.RetriesLeft.ToString(), "Information");
                            }
                            else
                            {
                                tweet.RetriesLeft = 0;

                                tweet.SendStatus = "Failed";

                                Trace.WriteLine("[Publish] Error publishing tweet " + tweet.RowKey + ". Out of retries.", "Information");

                                updateQueue.DeleteMessage(queueItem);
                            }
                            Repository.Update(tweet);
                        }
                    }
                }
                Thread.Sleep(60000);
            }
        }

        public override bool OnStart()
        {
            // Set the maximum number of concurrent connections 
            ServicePointManager.DefaultConnectionLimit = 12;

            DiagnosticMonitor.Start("DiagnosticsConnectionString");

            // For information on handling configuration changes
            // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.
            RoleEnvironment.Changing += RoleEnvironmentChanging;

            CloudStorageAccount.SetConfigurationSettingPublisher((configName, configSetter) =>
            {
                configSetter(RoleEnvironment.GetConfigurationSettingValue(configName));
            });

            return base.OnStart();
        }

        private void RoleEnvironmentChanging(object sender, RoleEnvironmentChangingEventArgs e)
        {
            // If a configuration setting is changing
            if (e.Changes.Any(change => change is RoleEnvironmentConfigurationSettingChange))
            {
                // Set e.Cancel to true to restart this role instance
                e.Cancel = true;
            }
        }
    }
}